package com.foreveross.fetl.queue;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Calendar;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import javax.inject.Inject;

import com.foreveross.fetl.common.HBaseTable;
import com.google.common.collect.Maps;
import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.net.ftp.FTPReply;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.log4j.Logger;
import org.apache.tools.zip.ZipOutputStream;
import org.springframework.stereotype.Component;

import com.common.remote.ftp.FtpConfig;
import com.common.remote.sftp.SFtpFileTransferClient;
import com.dayatang.domain.InstanceFactory;
import com.foreveross.fetl.application.ExcelInfoApplication;
import com.foreveross.fetl.application.ITaskTimeApplication;
import com.foreveross.fetl.common.cfg.system.ReportProperties;
import com.foreveross.fetl.common.hbase.client.HBaseHelper;
import com.foreveross.fetl.common.vo.FetlErrorMessage;
import com.foreveross.fetl.common.vo.hangxin.PlaneInfo;
import com.foreveross.fetl.common.vo.model.doub.DoublePlaneInfo;
import com.foreveross.fetl.queue.util.ExchangeRateUtil;
import com.foreveross.fetl.queue.util.HangxinDateUtil;
import com.foreveross.fetl.queue.util.LogMessageUtils;
import com.foreveross.fetl.util.AirLinesData;
import com.foreveross.fetl.util.DateUtils;
import com.foreveross.fetl.util.FetlErrorUtils;
import com.foreveross.fetl.util.HbaseToPlaneInfoUtil;
import com.foreveross.fetl.util.JsonUtil;
import com.foreveross.fetl.util.PlaneInfoUtil;
import com.foreveross.fetl.util.RowkeyUtils;
import com.foreveross.fetl.util.StringUtils;
import com.foreveross.fetl.util.ZipUtils;
import com.foreveross.fetl.vo.ExcelInfoVO;
import com.foreveross.fetl.vo.TaskTimeVO;
import com.sun.jersey.api.MessageException;

@Component
public class MetadataForAirsDoublePreprocess {
    private static Logger log = Logger.getLogger("metadataForAirs");

    @Inject
    private ExcelInfoApplication excelInfoApplication;

    private int tasksNumber;//任务数 多少条rowkey

    private int failuresNumber;//失败数

    private int insertNumber; //插入条数

    private long excuteStartTime;

    private String taskTime = null;

    private ZipOutputStream HXzipOut = null;
    private File HX;

    private StringBuffer keys = new StringBuffer();
    private StringBuffer results = new StringBuffer();

    private Boolean isThread = false;
    private int threadCount = 4;//线程数（查询HBASE和生成TXT文件时）

    /**
     * 初始化计数，添加日志
     */
    public void rollBackNumber() {
        tasksNumber = 0;
        failuresNumber = 0;
        insertNumber = 0;
        excuteStartTime = System.currentTimeMillis();
        keys.setLength(0);
        results.setLength(0);
        LogMessageUtils.create("航信元数据", this.getClass().getSimpleName());
    }

    public void execute() {
        Calendar calendar = Calendar.getInstance();
        int hour = calendar.get(Calendar.HOUR_OF_DAY);
        List<TaskTimeVO> listTask = InstanceFactory.getInstance(ITaskTimeApplication.class).find(null, FetlErrorUtils.metadataForAirs, hour);
        if (null != listTask && listTask.size() > 0) {
            rollBackNumber();
            Date sTime = null;
            Date eTime = null;
            String dateStr = DateUtils.format(new Date(), "yyyyMMdd");
            String start = null;
            String end = null, ends = null;
            ends = dateStr + (listTask.get(0).getGrabStop() < 10 ? "0" + listTask.get(0).getGrabStop() : listTask.get(0).getGrabStop());
            //创建航信zip路径
            try {
                HX = null;
                HXzipOut = null;
                HX = createHXZipFile(true, DateUtils.parse(ends, "yyyyMMddHH"));
                FileOutputStream HXfop = new FileOutputStream(HX);
                HXzipOut = new ZipOutputStream(HXfop);
                for (TaskTimeVO vo : listTask) {
                    if (vo.getGrabStart() < 10)
                        start = dateStr + "0" + vo.getGrabStart();
                    else
                        start = dateStr + vo.getGrabStart();
                    if (vo.getGrabStop() < 10)
                        end = dateStr + "0" + vo.getGrabStop();
                    else
                        end = dateStr + vo.getGrabStop();
                    sTime = DateUtils.parse(start, "yyyyMMddHH");
                    eTime = DateUtils.parse(end, "yyyyMMddHH");
                    List<String> lines = AirLinesData.getDoubleMapAirLines().get("hangxin").get(vo.getChannel());
                    Map<String, List<String>> hangxinMap = new HashMap<String, List<String>>();
                    hangxinMap.put(vo.getChannel(), lines);
                    Map<String, Map<String, List<String>>> map = new HashMap<String, Map<String, List<String>>>();
                    map.put("hangxin", hangxinMap);
                    process(sTime, eTime, map);
                }
                try {
                    log.error("关闭航信ZIP");
                    if (HXzipOut != null) {
                        HXzipOut.close();
                        HXzipOut.flush();
                    }
                    if (HXfop != null) {
                        HXfop.close();
                        HXfop.flush();
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            } catch (Exception e) {
                log.error("航信生成出错，请检查!");
                e.printStackTrace();
            }
            try {
                //执行完ZIP，关闭对应的流
                uploadFTP();
            } catch (Exception e) {
                log.error("航信上传出问题，请检查!");
                e.printStackTrace();
            }

            LogMessageUtils.build("航信元数据", this.getClass().getSimpleName(), 2, results.toString(),
                    failuresNumber, tasksNumber, insertNumber, taskTime, excuteStartTime, keys.toString());
            MetadataCountPreprocess count = new MetadataCountPreprocess();
            count.countforExcel(calendar.getTime());
        }
    }

    public void manualExecute(final String start, String time) {
        log.error("manualExcue 处理方法。。。 ");
        this.rollBackNumber();
        Map<String, Map<String, List<String>>> map = AirLinesData.getDoubleMapAirLines();
        try {
            List<String> timeList = Arrays.asList(time.split(";"));
            taskTime = start + " " + time;
            String str = start;
            str += timeList.get(0);
            Date startTime = DateUtils.parse(str, "yyyyMMddHH");
            Date endTime = DateUtils.parse(start + timeList.get(timeList.size() - 1), "yyyyMMddHH");

            //创建航信zip路径
            try {
                HX = null;
                HXzipOut = null;
                HX = createHXZipFile(true, endTime);
                FileOutputStream HXfop = new FileOutputStream(HX);
                HXzipOut = new ZipOutputStream(HXfop);
                this.process(startTime, endTime, map);

                try {
                    log.error("关闭航信ZIP");
                    if (HXfop != null) {
                        HXzipOut.close();
                        HXzipOut.flush();
                    }
                    if (HXfop != null) {
                        HXfop.close();
                        HXfop.flush();
                    }
                } catch (Exception e) {
                    log.error("关闭流出错！");
                    e.printStackTrace();
                }
                //执行完ZIP，关闭对应的流
                uploadFTP();
            } catch (Exception e) {
                log.error("航信上传出错!");
                e.printStackTrace();
            }
            LogMessageUtils.build("航信元数据", this.getClass().getName(), 2, results.toString(),
                    failuresNumber, tasksNumber, insertNumber, taskTime, excuteStartTime, keys.toString());
        } catch (Exception e) {
            log.error(StringUtils.getStackTrace(e));
            LogMessageUtils.build("航信元数据", this.getClass().getName(), 3, StringUtils.getStackTrace(e), failuresNumber, tasksNumber,
                    insertNumber, taskTime, excuteStartTime, keys.toString());
        }
    }


    public void hangxinExcute(String start, String end, String channel) {
        log.error("hangxinExcute 处理方法。。。 ");
        this.rollBackNumber();
        try {
            Date startTime = DateUtils.parse(start, "yyyyMMddHH");
            HXzipOut = null;
            HX = null;
            Date endTime = DateUtils.parse(end, "yyyyMMddHH");
            HX = createHXZipFile(true, endTime);
            FileOutputStream HXfop = new FileOutputStream(HX);
            HXzipOut = new ZipOutputStream(HXfop);
            if ("all".equals(channel)) {
                Map<String, Map<String, List<String>>> map = AirLinesData.getDoubleMapAirLines();
                process(startTime, endTime, map);
            } else {
                List<String> lines = AirLinesData.getDoubleMapAirLines().get("hangxin").get(channel);
                Map<String, List<String>> hangxinMap = new HashMap<String, List<String>>();
                hangxinMap.put(channel, lines);
                Map<String, Map<String, List<String>>> map = new HashMap<String, Map<String, List<String>>>();
                map.put("hangxin", hangxinMap);
                process(startTime, endTime, map);
            }
            try {
                log.error("关闭航信ZIP");
                if (HXfop != null) {
                    HXzipOut.close();
                    HXzipOut.flush();
                }
                if (HXfop != null) {
                    HXfop.close();
                    HXfop.flush();
                }
            } catch (Exception e) {
                log.error("关闭流出错！");
                e.printStackTrace();
            }
            //执行完ZIP，关闭对应的流
            uploadFTP();
            LogMessageUtils.build("航信元数据", this.getClass().getName(), 2, results.toString(),
                    failuresNumber, tasksNumber, insertNumber, taskTime, excuteStartTime, keys.toString());
        } catch (Exception e) {
            log.error(StringUtils.getStackTrace(e));
            LogMessageUtils.build("航信元数据", this.getClass().getName(), 3, StringUtils.getStackTrace(e), failuresNumber, tasksNumber,
                    insertNumber, taskTime, excuteStartTime, keys.toString());
        }

    }

    private void process(Date time, Date endTime, Map<String, Map<String, List<String>>> map) {
        log.error(DateUtils.format(time, "yyyyMMddHH"));
        log.error(DateUtils.format(endTime, "yyyyMMddHH"));
        //重新初始化汇率
        ExchangeRateUtil.initExchangeRate(DateUtils.format(new Date(), "yyyyMMdd"));

        /**
         * 根据航信新需求，一次只上传一个.zip文件
         * 在此之前生成所有的txt文件
         */
        //生成文件
        if (HX == null) {
            try {
                HXzipOut = null;
                HX = new File(ReportProperties.getFilePath() + "company/hengtuo_" + DateUtils.format(endTime, "yyyyMMddHH") + "00" + ".zip");
                HX.createNewFile();
                HXzipOut = new ZipOutputStream(new FileOutputStream(HX));
            } catch (IOException e) {
                log.error("航信路径创建失败！");
                throw new MessageException("航信路径创建失败，请联系管理员！");
            }
        }

        try {
            for (String companyName : map.keySet()) {
                Map<String, List<String>> airLines = map.get(companyName);
                for (String hbaseKey : airLines.keySet()) {

                    log.error("本次处理的渠道：" + hbaseKey);

                    ZipOutputStream LOCALzipOut = null;
                    FileOutputStream LOCALfop = null;
                    try {

                        Map<String, String> fileNames = new HashMap<String, String>();
                        tasksNumber++;
                        File localZIP = null;//本地.zip路径
                        localZIP = new File(ReportProperties.getFilePath() + "company/" + hbaseKey + "_" + DateUtils.format(endTime, "yyyy-MM-dd_HH") + ".zip");
                        if (!localZIP.exists()) {
                            try {
                                localZIP.createNewFile();
                            } catch (IOException e) {
                                log.error("创建本地FTP路径失败");
                                e.printStackTrace();
                            }
                        }
                        LOCALfop = new FileOutputStream(localZIP);
                        LOCALzipOut = new ZipOutputStream(LOCALfop);

                        if (isThread) {
                            //调用线程执行
                            ExecutorService service = Executors.newFixedThreadPool(threadCount);
                            service.execute(new FormHbaseThread(results, keys, HXzipOut, LOCALzipOut, hbaseKey, airLines.get(hbaseKey), time, endTime));
                            service.shutdown();
                            while (true) {
                                if (service.isTerminated()) {
                                    log.error("结束了");
                                    break;
                                }
                            }
                        } else {
                            int tsd = 0;

                            //处理国内去哪儿的数据。
                            if ("00002".equals(hbaseKey)) {//处理国内的对应的航线。
                                List<String> list = Arrays.asList("TAO-HRB", "HRB-TAO", "TAO-CSX", "CSX-TAO", "TAO-CTU", "CTU-TAO");
                                log.error("处理去哪儿的国内数据：" + list);
                                tsd = this.findFormHbaseDome(HXzipOut, LOCALzipOut, hbaseKey, list, time, endTime,tsd); // 创建文件
                            }

                            this.findFormHbase(HXzipOut, LOCALzipOut, hbaseKey, airLines.get(hbaseKey), time, endTime,tsd); // 创建文件

                        }
                        //关闭流
                        try {
                            if (LOCALzipOut != null) {
                                LOCALzipOut.close();
                                LOCALzipOut.flush();
                            }
                            if (LOCALfop != null) {
                                LOCALfop.close();
                                LOCALfop.flush();
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

                        //上传到本地服务器
                        fileNames.put(hbaseKey, localZIP.getPath());
                        for (String key : fileNames.keySet()) {
                            File file = null;
                            file = new File(fileNames.get(key));
                            log.error(String.format("上传到本地【%s】", key));
                            this.uploadFile(file, key, endTime, companyName);

                        }

                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

    }


    /**
     * 加载文件至.ZIP（压缩至.zip）
     *
     * @param file    需要加载的文件
     * @param zipFile zip包
     * @throws Exception
     */
    public void intoZIP(File file, ZipOutputStream zipFile) {
        try {
            log.error("txt文件加载至.zip文件");
            ZipUtils.zipFile(file, zipFile);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void uploadFTPHX(File uploadFile) {
        //上传到航信服务器
        log.error("上传到航信服务器...");
        boolean success = false;
        if (uploadFile != null) {
            try {
                // 上传文件到FTP
                success = this.uploadToHangxin(uploadFile);
                results.append(uploadFile.getName() + " " + uploadFile.length() + "<br>");
            } catch (Exception e) {
                success = this.uploadToHangxin(uploadFile);
            }
            if (success) log.error("已上传至航信服务器完成！");
            if (!success) {
                log.error("上传至航信服务器出错,请联系管理员！");
                results.append(uploadFile.getName() + " upload error" + "<br>");
            }
        }
    }

    /**
     * 创建.zip（航信）
     * 航信：hengtuo_YYYY-MM-DD_XX.zip   如：hengtuo_2015-01-08_09(XX).zip
     *
     * @return
     */
    public File createHXZipFile(boolean isHX, Date endTime) {
        File uploadFile = null;
        try {
            if (isHX) {
                log.error("创建航信ZIP文件");
                uploadFile = new File(ReportProperties.getFilePath() + "company/hengtuo_" + DateUtils.format(endTime, "yyyyMMddHH") + "00" + ".zip");
                if (uploadFile.exists()) {
                    uploadFile = new File(ReportProperties.getFilePath() + "company/hengtuo_" + DateUtils.format(endTime, "yyyyMMddHH") + "00(1)" + ".zip");
                    if (uploadFile.exists()) {
                        File files_ht = new File(ReportProperties.getFilePath() + "company/");
                        File[] list_ht = files_ht.listFiles();
                        List<String> strList = new ArrayList<String>();
                        for (int f = 0; f < list_ht.length; f++) {
                            String t = null;
                            if (list_ht[f].getName().contains("hengtuo_") && list_ht[f].getName().contains("(")) {
                                t = list_ht[f].getName().replaceAll("^.*\\(", "").replaceAll("\\).*", "");
                                if (t != null && !"".equals(t)) strList.add(t);
                            }
                        }
                        //排序
                        Collections.sort(strList);
                        String str = strList.get(strList.size() - 1);
                        int c = Integer.parseInt(str != null ? str : "0") + 1;
                        uploadFile = new File(ReportProperties.getFilePath() + "company/hengtuo_" + DateUtils.format(endTime, "yyyyMMddHH") + "00(" + c + ")" + ".zip");
                    }
                }
            }
            if (uploadFile != null)
                try {
                    if (!uploadFile.exists()) {
                        uploadFile.createNewFile();
                    }
                } catch (Exception e) {
                    log.error("zip文件创建失败！");
                    e.printStackTrace();
                }
            return uploadFile;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    public void uploadFTP() {
        uploadFTPHX(HX);
    }


    /**
     * 从hbase里读数据，并生成txt文件
     *
     * @param source
     * @param lines
     * @param time
     */
    private int findFormHbase(ZipOutputStream hxZIP, ZipOutputStream localZIP, String source, List<String> lines, Date time, Date endTime,int tsd) {
        int thousand = tsd;
        String path = ReportProperties.getFilePath() + "company/newFile/" + source + "/";
        Calendar cStart = Calendar.getInstance();
        cStart.setTime(time);
        Calendar cEnd = Calendar.getInstance();
        cEnd.setTime(endTime);
        int i = 1;
        int count = 0;
        String id = null;
        keys.append("渠道： " + source);
        StringBuilder josnString = new StringBuilder(512 * 1024);
        String separator = System.getProperty("line.separator");
        List<String> listkeys = new ArrayList<String>();
        Set<Date> flightDateSet = HangxinDateUtil.getFlightDateSet();
        Integer[] returnDays = HangxinDateUtil.getReturnDays();
        int flag = 0;
        if (lines.size() > 0) {
            while (cStart.before(cEnd)) {
                String timeStr = DateUtils.format(cStart.getTime(), "yyyyMMddHH");
                for (String line : lines) {
                    String[] citys = line.split("-");
                    for (Date flightDate : flightDateSet) {
                        String goDateStr = DateUtils.format(flightDate, "yyyyMMdd");
                        for (int j = 0; j < returnDays.length; j++) {
                            ResultScanner rs = null;
                            Calendar c = Calendar.getInstance();
                            c.setTime(flightDate);
                            c.add(Calendar.DAY_OF_MONTH, returnDays[j]);
                            Date returnFlightDate = c.getTime();
                            String returnDay = DateUtils.format(returnFlightDate, "yyyyMMdd");
                            String startRow = RowkeyUtils.getRowKey(timeStr, source, citys[0], citys[1], goDateStr, returnDay, RowkeyUtils.START_ROW_KEY);
                            String endRow = RowkeyUtils.getRowKey(timeStr, source, citys[0], citys[1], goDateStr, returnDay, RowkeyUtils.STOP_ROW_KEY);
                            try {
                                //查询Hbase
                                try {
                                    rs = HBaseHelper.getRecords("CRAWL_PLANE_FREIGHT_DOUBLE",
                                            startRow, endRow, new String[]{"planeinfo", "transitsInfo", "cabinsInfo", "stopOversInfo",
                                                    "returnPlaneInfosInfo", "returnCabinsInfo", "returnStopOversInfo", "returnTransitsInfo"}, null);
                                    rs.iterator();
                                } catch (Exception e) {
                                    rs = HBaseHelper.getRecords("CRAWL_PLANE_FREIGHT_DOUBLE",
                                            startRow, endRow, new String[]{"planeinfo", "transitsInfo", "cabinsInfo", "stopOversInfo",
                                                    "returnPlaneInfosInfo", "returnCabinsInfo", "returnStopOversInfo", "returnTransitsInfo"}, null);
                                    //如果从hbase里读数据失败了就再来一次
                                }
                                if (rs != null) {
                                    for (Result res : rs) {
                                        try {
                                            Map<String, Map<String, String>> curMap = HBaseHelper
                                                    .transformHBaseRowToMap(res);
                                            Map<String, String> curPalneInfoMap = curMap.get("planeinfo");

                                            id = new String(res.getRow());
                                            listkeys.add(id);
                                            if (i == 1) {
                                                keys.append(" 开始rowkey： " + id + "<br>");
                                                i++;
                                            }
                                            DoublePlaneInfo planeInfo = null;
                                            double refePrice = ExchangeRateUtil.getRefePrice(curPalneInfoMap.get("currency"));
                                            if (curPalneInfoMap.get("grabChannelNum").equals("00006") && null == curPalneInfoMap.get("currency")) {
                                                refePrice = ExchangeRateUtil.getRefePrice("USD");
                                            }
                                            planeInfo = HbaseToPlaneInfoUtil.hbaseToPlaneInfo(curMap, refePrice, DoublePlaneInfo.class);
                                            if (null != planeInfo && null != planeInfo.getFlightNo()) {
                                                List<PlaneInfo> list2 = PlaneInfoUtil.changeToPlaneInfo(planeInfo);
                                                if (null != list2 && list2.size() > 0) {
                                                    for (PlaneInfo vo : list2) {
                                                        josnString.append(JsonUtil.getJSONString(vo));
                                                        josnString.append(separator);
                                                        count++;
                                                    }
                                                }
                                            }
                                            if (count >= 10000) {//存在数据，则生成txt文件
                                                thousand++;
                                                count = 0;
                                                flag++;
                                                String fileName = source + "_" + DateUtils.format(endTime, "yyyy-MM-dd_HH") + "_" + thousand + ".txt";
                                                ZipUtils.creatFile(path, fileName, josnString.toString());
                                                josnString.delete(0, josnString.length());
                                                //生成的txt文件加载到.zip
                                                //加载到本地
                                                File file = new File(path + "/" + fileName);
                                                intoZIP(file, localZIP);
                                                intoZIP(file, hxZIP);
                                                new File(path + "/" + fileName).delete();
                                                listkeys.clear();
                                                count = 0;
                                            }

                                        } catch (Exception e) {
                                            FetlErrorMessage vo = FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.metadataForAirs,
                                                    this.getClass().getName(), DateUtils.format(endTime, "yyyyMMddHH"), e);
                                            vo.setKeys(listkeys);
                                            FetlErrorUtils.saveToHbase(vo);
                                            listkeys.clear();
                                            count = 0;
                                        }
                                    }
                                }
                            } catch (Exception e) {
                                log.error(StringUtils.getStackTrace(e));
                                FetlErrorMessage vo = FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.metadataForAirs,
                                        this.getClass().getName(), DateUtils.format(endTime, "yyyyMMddHH"), e);
                                vo.setStartRowKey(startRow);
                                vo.setEndRowkey(endRow);
                                FetlErrorUtils.saveToHbase(vo);
                            } finally {
                                if (rs != null) rs.close();
                                rs = null;
                            }
                        }
                    }
                }
                cStart.add(Calendar.HOUR_OF_DAY, 1);// 开始时间加一小时
            }
        }
        try {
            String fileName = null;
            if (count > 0) {
                //无论有无数据都创建路径
                thousand++;
                fileName = source + "_" + DateUtils.format(endTime, "yyyy-MM-dd_HH") + "_" + thousand + ".txt";
                ZipUtils.creatFile(path, fileName, josnString.toString());
                josnString.delete(0, josnString.length());
                File file = new File(path + "/" + fileName);
                intoZIP(file, localZIP);
                intoZIP(file, hxZIP);
                new File(path + "/" + fileName).delete();
                listkeys.clear();
            }
            if (lines.size() == 0) {
                thousand++;
                fileName = source + "_" + DateUtils.format(endTime, "yyyy-MM-dd_HH") + "_" + thousand + ".txt";
                ZipUtils.creatFile(path, fileName, josnString.toString());
                josnString.delete(0, josnString.length());
                File file = new File(path + "/" + fileName);
                intoZIP(file, localZIP);
                intoZIP(file, hxZIP);
                new File(path + "/" + fileName).delete();
                listkeys.clear();
            }
            if (flag == 0 && lines.size() > 0) {
                thousand++;
                fileName = source + "_" + DateUtils.format(endTime, "yyyy-MM-dd_HH") + "_" + thousand + ".txt";
                ZipUtils.creatFile(path, fileName, josnString.toString());
                josnString.delete(0, josnString.length());
                File file = new File(path);
                intoZIP(file, localZIP);
                intoZIP(file, hxZIP);
                new File(path + "/" + fileName).delete();
                listkeys.clear();
            }

        } catch (Exception e) {
            FetlErrorMessage vo = FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.metadataForAirs,
                    this.getClass().getName(), DateUtils.format(endTime, "yyyyMMddHH"), e);
            vo.setKeys(listkeys);
            FetlErrorUtils.saveToHbase(vo);
            listkeys.clear();
        }
        if (i > 1) {
            keys.append(" 结束rowkey： " + id + "<br>");
        } else {
            keys.append(" 没数据 <br> ");
        }
        return thousand;
    }

    /**
     * 从hbase里读数据，并生成txt文件
     *
     * @param source
     * @param lines
     * @param time
     */
    private int findFormHbaseDome(ZipOutputStream hxZIP, ZipOutputStream localZIP, String source, List<String> lines, Date time, Date endTime,int tsd) {
        log.error("读取航信国内数据，渠道为：" + source);

        int thousand = tsd;
        String path = ReportProperties.getFilePath() + "company/newFile/" + source + "/";

        Calendar cStart = Calendar.getInstance();
        cStart.setTime(time);
        Calendar cEnd = Calendar.getInstance();
        cEnd.setTime(endTime);

        int i = 1;
        int count = 0;
        String id = null;
        keys.append("渠道： " + source);
        StringBuilder josnString = new StringBuilder(512 * 1024);
        String separator = System.getProperty("line.separator");
        List<String> listkeys = new ArrayList<String>();

        int flag = 0;
        if (lines.size() > 0) {
            while (cStart.before(cEnd)) {
                String timeStr = DateUtils.format(cStart.getTime(), "yyyyMMddHH");
                for (String line : lines) {
                    String[] citys = line.split("-");
                    for (int j = 0; j < 45; j++) {// 处理往后45天的数据。
                        Calendar c = Calendar.getInstance();
                        c.add(Calendar.DAY_OF_MONTH, j);
                        String goDateStr = DateUtils.format(c.getTime(), "yyyyMMdd");

                        ResultScanner rs = null;
                        String startRow = RowkeyUtils.getRowKey(timeStr, source, citys[0], citys[1], goDateStr, RowkeyUtils.START_ROW_KEY);
                        String endRow = RowkeyUtils.getRowKey(timeStr, source, citys[0], citys[1], goDateStr,RowkeyUtils.STOP_ROW_KEY);
                        try {
                            //查询Hbase
                            try {
                                rs = HBaseHelper.getRecords(HBaseTable.BASE_DATA_FLIGHT,
                                        startRow, endRow, new String[]{"planeinfo", "transitsInfo", "cabinsInfo", "stopOversInfo"}, null);
                                rs.iterator();
                            } catch (Exception e) {
                                rs = HBaseHelper.getRecords(HBaseTable.BASE_DATA_FLIGHT,
                                        startRow, endRow, new String[]{"planeinfo", "transitsInfo", "cabinsInfo", "stopOversInfo"}, null);
                                //如果从hbase里读数据失败了就再来一次
                            }

                            if (rs != null) {
                                for (Result res : rs) {
                                    try {
                                        Map<String, Map<String, String>> curMap = HBaseHelper
                                                .transformHBaseRowToMap(res);
                                        Map<String, String> curPalneInfoMap = curMap.get("planeinfo");

                                        id = new String(res.getRow());
                                        listkeys.add(id);
                                        if (i == 1) {
                                            keys.append(" 开始rowkey： " + id + "<br>");
                                            i++;
                                        }
                                        DoublePlaneInfo planeInfo = null;
                                        double refePrice = ExchangeRateUtil.getRefePrice(curPalneInfoMap.get("currency"));
                                        if (curPalneInfoMap.get("grabChannelNum").equals("00006") && null == curPalneInfoMap.get("currency")) {
                                            refePrice = ExchangeRateUtil.getRefePrice("USD");
                                        }
                                        planeInfo = HbaseToPlaneInfoUtil.hbaseToPlaneInfo(curMap, refePrice, DoublePlaneInfo.class);
                                        if (null != planeInfo && null != planeInfo.getFlightNo()) {
                                            List<PlaneInfo> list2 = PlaneInfoUtil.changeToPlaneInfo(planeInfo);
                                            if (null != list2 && list2.size() > 0) {
                                                for (PlaneInfo vo : list2) {
                                                    josnString.append(JsonUtil.getJSONString(vo));
                                                    josnString.append(separator);
                                                    count++;
                                                }
                                            }
                                        }
                                        if (count >= 10000) {//存在数据，则生成txt文件
                                            thousand++;
                                            count = 0;
                                            flag++;
                                            String fileName = source + "_" + DateUtils.format(endTime, "yyyy-MM-dd_HH") + "_" + thousand + ".txt";
                                            ZipUtils.creatFile(path, fileName, josnString.toString());
                                            josnString.delete(0, josnString.length());
                                            //生成的txt文件加载到.zip
                                            //加载到本地
                                            File file = new File(path + "/" + fileName);
                                            intoZIP(file, localZIP);
                                            intoZIP(file, hxZIP);
                                            new File(path + "/" + fileName).delete();
                                            listkeys.clear();
                                            count = 0;
                                        }

                                    } catch (Exception e) {
                                        FetlErrorMessage vo = FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.metadataForAirs,
                                                this.getClass().getName(), DateUtils.format(endTime, "yyyyMMddHH"), e);
                                        vo.setKeys(listkeys);
                                        FetlErrorUtils.saveToHbase(vo);
                                        listkeys.clear();
                                        count = 0;
                                    }
                                }
                            }
                        } catch (Exception e) {
                            log.error(StringUtils.getStackTrace(e));
                            FetlErrorMessage vo = FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.metadataForAirs,
                                    this.getClass().getName(), DateUtils.format(endTime, "yyyyMMddHH"), e);
                            vo.setStartRowKey(startRow);
                            vo.setEndRowkey(endRow);
                            FetlErrorUtils.saveToHbase(vo);
                        } finally {
                            if (rs != null) rs.close();
                            rs = null;
                        }
                    }
                }
                cStart.add(Calendar.HOUR_OF_DAY, 1);// 开始时间加一小时
            }

        }
        try {
            String fileName = null;
            if (count > 0) {
                //无论有无数据都创建路径
                thousand++;
                fileName = source + "_" + DateUtils.format(endTime, "yyyy-MM-dd_HH") + "_" + thousand + ".txt";
                ZipUtils.creatFile(path, fileName, josnString.toString());
                josnString.delete(0, josnString.length());
                File file = new File(path + "/" + fileName);
                intoZIP(file, localZIP);
                intoZIP(file, hxZIP);
                new File(path + "/" + fileName).delete();
                listkeys.clear();
            }
            if (lines.size() == 0) {
                thousand++;
                fileName = source + "_" + DateUtils.format(endTime, "yyyy-MM-dd_HH") + "_" + thousand + ".txt";
                ZipUtils.creatFile(path, fileName, josnString.toString());
                josnString.delete(0, josnString.length());
                File file = new File(path + "/" + fileName);
                intoZIP(file, localZIP);
                intoZIP(file, hxZIP);
                new File(path + "/" + fileName).delete();
                listkeys.clear();
            }
            if (flag == 0 && lines.size() > 0) {
                thousand++;
                fileName = source + "_" + DateUtils.format(endTime, "yyyy-MM-dd_HH") + "_" + thousand + ".txt";
                ZipUtils.creatFile(path, fileName, josnString.toString());
                josnString.delete(0, josnString.length());
                File file = new File(path);
                intoZIP(file, localZIP);
                intoZIP(file, hxZIP);
                new File(path + "/" + fileName).delete();
                listkeys.clear();
            }


        } catch (Exception e) {
            FetlErrorMessage vo = FetlErrorMessage.createFetlErrorMessage(FetlErrorUtils.metadataForAirs,
                    this.getClass().getName(), DateUtils.format(endTime, "yyyyMMddHH"), e);
            vo.setKeys(listkeys);
            FetlErrorUtils.saveToHbase(vo);
            listkeys.clear();
        }
        if (i > 1) {
            keys.append(" 结束rowkey： " + id + "<br>");
        } else {
            keys.append(" 没数据 <br> ");
        }
        return thousand;
    }

    /**
     * 将文件上传到Hangxin FTP 上
     *
     * @param f 文件
     * @return
     */
    private boolean uploadToHangxin(File f) {
        String address = ReportProperties.getHangxinAddress();
        int port = Integer.parseInt(null == ReportProperties.getHangxinPort() ? "21" : ReportProperties.getHangxinPort());
        String userName = ReportProperties.getHangxinUserName();
        String passWord = ReportProperties.getHangxinPassword();
        String path = ReportProperties.getHangxinPath();
        if (!path.endsWith("/")) path += "/";
        boolean success = false;
        try {
            Long times = System.currentTimeMillis();
            if ("/".equals(path)) {
                success = this.uploadFile(address, port, userName, passWord, null, f);
            } else {
                success = this.uploadFile(address, port, userName, passWord, path, f);
            }
            log.error(f.getName());
            log.error(DateUtils.timeInMillis2Time(System.currentTimeMillis() - times));
        } catch (Exception e) {
            try {
                throw e;
            } catch (Exception e1) {
                e1.printStackTrace();
            }
        } finally {

        }
        return success;
    }

    /**
     * 将文件上传到 我们的SFTP上
     *
     * @param f
     * @param hbasekey
     * @param time
     * @param type
     */
    private void uploadFile(File f, String hbasekey, Date time, String type) {
        String address = ReportProperties.getAddress();
        int port = Integer.parseInt(null == ReportProperties.getPort() ? "22" : ReportProperties.getPort());
        String userName = ReportProperties.getUserName();
        String passWord = ReportProperties.getPassword();
        String path = ReportProperties.getPath();
        if (!path.endsWith("/")) path += "/";
        String dateType = DateUtils.format(time, "yyyyMMddHH");
        path = path.replace("webdata/", "");
        String dir = path + type + "/round-trip/" + dateType.substring(0, 4) + "/" + dateType.substring(4, 6) + "/" + dateType.substring(6, 8) + "/" + hbasekey;
        //ftp连接配置
        FtpConfig config = new FtpConfig(address, port, userName, passWord);
        SFtpFileTransferClient sftpClient = new SFtpFileTransferClient(config);
        InputStream in = null;
        boolean flag = false;
        try {
            try {
                log.error("链接ftp服务器");
                flag = sftpClient.connect();
            } catch (Exception e) {
                log.error(StringUtils.getStackTrace(e));
                Thread.sleep(10 * 1000L);
                flag = sftpClient.connect();
            }

            if (flag) {
                //如果目录不存在则创建
                String[] dirs = dir.split("/");
                StringBuffer buffer = new StringBuffer();
                for (String d : dirs) {
                    buffer.append(d + "/");
                    if (!sftpClient.isDirectoryExists(buffer.toString())) {
                        sftpClient.makeDirectory(buffer.toString());
                    }
                }
                in = new FileInputStream(f);
                log.error("上传至本地ftp");
                sftpClient.upload(in, dir + "/" + f.getName());

                sftpClient.disconnect();
                insertNumber++;
                ExcelInfoVO oldVo = excelInfoApplication.findByName(f.getName(), type, 2);
                if (null != oldVo) {
                    excelInfoApplication.delete(oldVo);
                }
                ExcelInfoVO excelInfoVO = new ExcelInfoVO();
                excelInfoVO.setCode(hbasekey);
                excelInfoVO.setDateType(dateType);
                excelInfoVO.setType(2);
                excelInfoVO.setFileType(type);
                excelInfoVO.setName(f.getName());
                excelInfoVO.setPath(f.getName());
                //	excelInfoVO.setPath(type+"/round-trip/"+dateType.substring(0, 4)+"/"+dateType.substring(4, 6)+"/"+dateType.substring(6, 8)+"/"+hbasekey+"/"+f.getName());
                excelInfoVO.setUploadDate(new Date());
                excelInfoApplication.saveOrUpdate(excelInfoVO);
                //		in.close();
            }
        } catch (Exception e) {
            try {
                throw e;
            } catch (Exception e1) {
                e1.printStackTrace();
            }
        } finally {
            if (null != sftpClient) {
                sftpClient.disconnect();
            }
            try {
                if (in != null) {
                    in.close();
                }
            } catch (Exception e) {
            }
        }
    }


    /**
     * Description: 向FTP服务器上传文件
     *
     * @param url      FTP服务器hostname
     * @param port     FTP服务器端口
     * @param username FTP登录账号
     * @param password FTP登录密码
     * @param path     FTP服务器保存目录
     * @param filename 上传到FTP服务器上的文件名
     * @param input    输入流
     * @return 成功返回true，否则返回false
     */
    private boolean uploadFile(String url, int port, String username, String password, String path, File file) {
        boolean success = false;
        FTPClient ftp = new FTPClient();
        InputStream input = null;
        try {
            int reply;
            boolean flag = false;
            String flagStr = "";
            input = new FileInputStream(file);//hengtuo%yyyymmdd0900.dat
            try {
                String nameStr = file.getName().replace("_", "%");
                if (nameStr.indexOf("(") != -1) {//存在
                    flagStr = nameStr.split("\\(")[0] + ".dat";
                } else {
                    flagStr = nameStr.split("\\.")[0] + ".dat";
                }
                ftp.connect(url, port);//连接FTP服务器
                //如果采用默认端口，可以使用ftp.connect(url)的方式直接连接FTP服务器
                ftp.login(username, password);//登录
                if (!flag) {
                    ftp.connect(url, port);
                    ftp.login(username, password);
                }
            } catch (Exception e) {
                ftp.connect(url, port);//连接FTP服务器
                //如果采用默认端口，可以使用ftp.connect(url)的方式直接连接FTP服务器
                ftp.login(username, password);//登录
                if (!flag) {
                    ftp.connect(url, port);
                    ftp.login(username, password);
                }
            }
            ftp.enterLocalPassiveMode();//设置客户端PASV模式 (客户端主动连服务器端；端口用20) 有防火墙用户不能使用主动模式 ，这是因为防火墙不允许来自网外的主动连接，所以用户必须同使用被动模 式
            ftp.setFileType(FTPClient.BINARY_FILE_TYPE);
            ftp.setBufferSize(1024 * 1024 * 10);
            reply = ftp.getReplyCode();
            if (!FTPReply.isPositiveCompletion(reply)) {
                ftp.disconnect();
                return success;
            }
            ftp.changeWorkingDirectory(path);
            ftp.storeFile(file.getName(), input);
            input.close();
//			File flagFile=new File(file.getName().replace(".zip", "_flag"));
            File flagFile = new File(flagStr);//hengtuo%yyyymmdd0900.dat
            if (!flagFile.exists()) {
                flagFile.createNewFile();
            }
            input = new FileInputStream(flagFile);
            ftp.storeFile(flagFile.getName(), input);
            input.close();
            flagFile.delete();
            success = true;
        } catch (IOException e) {
            //log.error(StringUtils.getStackTrace(e));
            try {
                throw e;
            } catch (IOException e1) {
                e1.printStackTrace();
            }
        } finally {
            try {
                if (null != input)
                    input.close();
            } catch (IOException e) {
                log.error(StringUtils.getStackTrace(e));
            }
            try {
                ftp.logout();
            } catch (IOException e) {
                log.error(StringUtils.getStackTrace(e));
            }
            if (ftp.isConnected()) {
                try {
                    ftp.disconnect();
                } catch (IOException ioe) {
                    log.error(StringUtils.getStackTrace(ioe));
                }
            }
        }
        return success;
    }
}
